package com.bbx.flink.demo.jdbc_sink;

import com.bbx.flink.demo.entity.Books;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class JDBCDemo {


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Books books1 = new Books();
        books1.setId(1);
        books1.setAuthor("sss");
        books1.setTitle("sss");
        books1.setPrice(22.45);
        books1.setQty(1);


        env.fromElements(books1)
                .addSink(JdbcSink.sink(
                        "insert into books (id,title,author,price,qty)values(?,?,?,?,?) ",
                        (ps, t) -> {
                            ps.setInt(1, t.getId());
                            ps.setString(2, t.getTitle());
                            ps.setString(3, t.getAuthor());
                            ps.setDouble(4, t.getPrice());
                            ps.setInt(5, t.getQty());
                        },
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withUrl("jdbc:mysql://192.168.10.7:3306/flink_jdbc?characterEncoding=UTF-8&serverTimezone=GMT%2B8")
                                .withDriverName("com.mysql.cj.jdbc.Driver")
                                .withUsername("root")
                                .withPassword("123456")
                                .build()
                ));

        env.execute();

    }
}
